package org.digitarts.rtms.spring.collectors.jms;

import java.util.HashSet;
import java.util.Set;

import javax.jms.Message;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import org.digitarts.rtms.helpers.StringHelper;
import org.digitarts.rtms.spring.collectors.CollectorBase;

public class JMSCollector extends CollectorBase
{

	protected QueueConnectionFactory		queueConnectionFactory	= null;
	protected TopicConnectionFactory		topicConnectionFactory	= null;
	protected QueueConnection				queueConnection			= null;
	protected TopicConnection				topicConnection			= null;
	protected QueueSession					queueSession			= null;
	protected TopicSession					topicSession			= null;
	protected QueueSender					queueSender				= null;
	protected TopicPublisher				topicPublisher			= null;
	protected QueueReceiver					queueReceiver			= null;
	protected Queue							temporaryQueue			= null;
	protected Topic							temporaryTopic			= null;
	protected TopicSubscriber				topicSubscriber			= null;
	protected boolean						testTopic				= false;
	protected boolean						testQueue				= false;
	protected long							messageTimeOut			= 5000;
	protected long							ttlAdd					= 500;
	protected int							priority				= Message.DEFAULT_PRIORITY;
	protected int							deliveryMode			= Message.DEFAULT_DELIVERY_MODE;
	protected Set<ISyntehticMessageFactory>	messageFactories		= new HashSet<ISyntehticMessageFactory>();
	protected QueueBrowser					queueBrowser			= null;

	// protected Set<Queue> targetQueues = new HashSet<Queue>();

	public JMSCollector()
	{
		super();
	}

	@Override
	public void initCallback()
	{
		testQueue = (queueConnectionFactory != null);
		testTopic = (topicConnectionFactory != null);
		if (testQueue)
		{
			if (log.isDebugEnabled())
				log.debug("Initializing Queue Testing...");
			try
			{
				queueConnection = queueConnectionFactory.createQueueConnection();
				queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
				temporaryQueue = queueSession.createTemporaryQueue();
				queueSender = queueSession.createSender(temporaryQueue);
				queueReceiver = queueSession.createReceiver(temporaryQueue);
				queueBrowser = queueSession.createBrowser(temporaryQueue);
				queueConnection.start();
			}
			catch (Exception e)
			{
				testQueue = false;
				cleanupQueues();
				log.error("Failed to initialize queue testing", e);
			}
			finally
			{
			}
		}
	}

	@Override
	public void collectCallback() throws Exception
	{
		try
		{
			long start = System.currentTimeMillis();
			if (testQueue || testTopic)
			{
				for (ISyntehticMessageFactory factory : messageFactories)
				{
					if (testQueue)
					{
						Message[] testMessages = factory.createMessage(queueSession);
						if (log.isDebugEnabled())
							log.debug("Issuing Test with " + testMessages.length + " " + factory.getName() + " Messages");
						long sendElapsed = 0;
						long receiveElapsed = 0;
						long totalElapsed = 0;
						int timeOuts = 0;
						for (Message testMessage : testMessages)
						{
							long sendStart = System.currentTimeMillis();
							queueSender.send(testMessage, deliveryMode, priority, messageTimeOut + ttlAdd);
							sendElapsed += System.currentTimeMillis() - sendStart;
							long receiveStart = System.currentTimeMillis();
							Message testMessage2 = queueReceiver.receive(messageTimeOut);
							receiveElapsed += System.currentTimeMillis() - receiveStart;
							totalElapsed += System.currentTimeMillis() - sendStart;
							// if(log.isDebugEnabled())log.debug("SyntheticMessage Test On ["
							// + factory.getName() +
							// "] with total elapsed time:" + totalElapsed +
							// " ms.");
							if (testMessage2 == null)
							{
								timeOuts++;
							}
						}
						if (timeOuts > 0)
						{
							tracer.traceIncident(timeOuts, StringHelper.append(tracingNameSpace, factory.getName(), "Time Outs"));
						}
						tracer.traceSticky(receiveElapsed, StringHelper.append(tracingNameSpace, factory.getName(), "Receive Time"));
						tracer.traceSticky(sendElapsed, StringHelper.append(tracingNameSpace, factory.getName(), "Send Time"));
						tracer.traceSticky(totalElapsed, StringHelper.append(tracingNameSpace, factory.getName(), "Round Trip Time"));

					}
				}
			}
			long elapsed = System.currentTimeMillis() - start;
			tracer.trace(elapsed, elapsedTimeMetricNameSpace);
		}
		catch (Exception e)
		{
			log.error("JMS Collection failed", e);
		}
	}

	@Override
	public void stopCallback()
	{
		cleanupQueues();
	}

	protected void cleanupQueues()
	{
		try
		{
			queueReceiver.close();
		}
		catch (Exception e)
		{
		}
		try
		{
			queueSender.close();
		}
		catch (Exception e)
		{
		}
		try
		{
			queueSession.close();
		}
		catch (Exception e)
		{
		}
		try
		{
			queueConnection.stop();
		}
		catch (Exception e)
		{
		}
		try
		{
			queueConnection.close();
		}
		catch (Exception e)
		{
		}
	}

	public QueueConnectionFactory getQueueConnectionFactory()
	{
		return queueConnectionFactory;
	}

	public void setQueueConnectionFactory(QueueConnectionFactory queueConnectionFactory)
	{
		this.queueConnectionFactory = queueConnectionFactory;
	}

	public TopicConnectionFactory getTopicConnectionFactory()
	{
		return topicConnectionFactory;
	}

	public void setTopicConnectionFactory(TopicConnectionFactory topicConnectionFactory)
	{
		this.topicConnectionFactory = topicConnectionFactory;
	}

	public boolean isTestTopic()
	{
		return testTopic;
	}

	public void setTestTopic(boolean testTopic)
	{
		this.testTopic = testTopic;
	}

	public boolean isTestQueue()
	{
		return testQueue;
	}

	public void setTestQueue(boolean testQueue)
	{
		this.testQueue = testQueue;
	}

	public long getMessageTimeOut()
	{
		return messageTimeOut;
	}

	public void setMessageTimeOut(long messageTimeOut)
	{
		this.messageTimeOut = messageTimeOut;
	}

	public void setMessageFactories(Set<ISyntehticMessageFactory> messageFactories)
	{
		this.messageFactories = messageFactories;
	}

	public long getTtlAdd()
	{
		return ttlAdd;
	}

	public void setTtlAdd(long ttlAdd)
	{
		this.ttlAdd = ttlAdd;
	}

	public int getPriority()
	{
		return priority;
	}

	public void setPriority(int priority)
	{
		this.priority = priority;
	}

	public int getDeliveryMode()
	{
		return deliveryMode;
	}

	public void setDeliveryMode(int deliveryMode)
	{
		this.deliveryMode = deliveryMode;
	}

}
